package cn.doitedu.rtmk.demo4;

import cn.doitedu.rtmk.common.EventBean;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;


/**
 * 规则id:  rule-2：
 * 静态画像条件: active_level = 3 , gender = female
 * 实时动态画像标签条件: (k, b ,c)序列发生次数 >= 1
 * 触发条件: m行为且properties['p1']='v1'
 */
public class Rule_2_Calculator implements RuleCalculator{
    Table userProfileTable;
    JSONObject jsonObject;

    ValueState<Integer> rule2State;
    String[] rule2ActionSequence;
    @Override
    public void init(RuntimeContext runtimeContext) throws IOException {

        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "doitedu:2181");

        Connection hbaseConnection = ConnectionFactory.createConnection(conf);
        userProfileTable = hbaseConnection.getTable(TableName.valueOf("user_profile"));

        jsonObject = new JSONObject();


        // 申请一个keyedState，来记录rule-2中的用户行为序列条件中，用户完成到了第几步
        rule2State = runtimeContext.getState(new ValueStateDescriptor<Integer>("rule-2-state", Integer.class));

        rule2ActionSequence = new String[]{"k", "b", "c"};
    }

    @Override
    public void calc(EventBean eventBean, Collector<String> collector) throws IOException {

        String eventId = eventBean.getEventId();

        int currentStep = rule2State.value() == null ? 0 : rule2State.value();
        if (currentStep < 3 && eventId.equals(rule2ActionSequence[currentStep])) {
            rule2State.update(currentStep + 1);
        }

        // 判断事件 bean中的 eventId=要求的触发事件
        if (eventBean.getEventId().equals("m") && eventBean.getProperties().getOrDefault("p1", "").equals("v1")) {

            // 判断动态画像条件是否已满足
            if (rule2State.value() != null && rule2State.value() >= 3) {

                // 指定查询条件的行健
                Get getParam = new Get(Bytes.toBytes(eventBean.getUid()));
                // 指定要查询的标签（qualifier）
                getParam.addColumn(Bytes.toBytes("f"), Bytes.toBytes("active_level"));
                getParam.addColumn(Bytes.toBytes("f"), Bytes.toBytes("gender"));


                // 调用客户端查询
                Result result = userProfileTable.get(getParam);
                byte[] activeLevelBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("active_level"));
                byte[] genderBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("gender"));
                int activeLevel = Integer.parseInt(Bytes.toString(activeLevelBytes));
                String genderStr = Bytes.toString(genderBytes);


                // 判断该用户的画像标签值，是否满足规则要求的值
                if (activeLevel == 3 && "female".equals(genderStr)) {

                    jsonObject.put("uid", eventBean.getUid());
                    jsonObject.put("timestamp", eventBean.getTimestamp());
                    jsonObject.put("rule_id", "rule-002");

                    collector.collect(jsonObject.toJSONString());

                }

            }
        }

    }
}
